package com.flink.es;

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.connector.format.DecodingFormat;
import org.apache.flink.table.connector.format.EncodingFormat;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.factories.*;
import org.apache.flink.table.types.DataType;

import java.util.HashSet;
import java.util.Set;

/**
 * @Author hujf
 * @Date 2022/8/29 9:36
 * @Description mysql
 */
public class EsDynamicTableFactory implements DynamicTableSourceFactory, DynamicTableSinkFactory {


    public static final String IDENTIFIER = "http";
    public static final ConfigOption<String> URL = ConfigOptions.key("url")
            .stringType().noDefaultValue().withDescription("the jdbc database url.");
    public static final ConfigOption<String> HEADERS = ConfigOptions.key("headers")
            .stringType().noDefaultValue().withDescription("the http header.");
    private static final ConfigOption<String> BODY = ConfigOptions.key("body")
            .stringType().noDefaultValue().withDescription("the http body params.");
    private static final ConfigOption<String> TYPE = ConfigOptions.key("type")
            .stringType().noDefaultValue().withDescription("the http type.");

    private static final ConfigOption<String> FORMAT = ConfigOptions.key("format")
            .stringType().noDefaultValue().withDescription("the http format.");

    private static final ConfigOption<String> UNION_TYPE = ConfigOptions.key("union_type")
            .stringType().noDefaultValue().withDescription("the http format.");
    private static final ConfigOption<String> UNION_URL = ConfigOptions.key("union_url")
            .stringType().noDefaultValue().withDescription("the http format.");
    private static final ConfigOption<String> UNION_DATA = ConfigOptions.key("union_data")
            .stringType().noDefaultValue().withDescription("the http format.");

    @Override
    public DynamicTableSink createDynamicTableSink(Context context) {
//        FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
//        ReadableConfig config = helper.getOptions();
//        helper.validate();
//
//        String url = config.get(URL);
//        String type = config.get(TYPE);
//        String header = config.get(HEADERS);
//        String body = config.get(BODY);
//
//        // discover a suitable encoding format
//        final EncodingFormat<SerializationSchema<RowData>> encodingFormat = helper.discoverEncodingFormat(
//                SerializationFormatFactory.class,
//                FactoryUtil.FORMAT);
//
//        // derive the produced data type (excluding computed columns) from the catalog table
//        final DataType producedDataType = context.getCatalogTable().getSchema().toPhysicalRowDataType();
//        TableSchema tableSchema = context.getCatalogTable().getSchema();

        return (DynamicTableSink) new HttpSink();
    }

    @Override
    public DynamicTableSource createDynamicTableSource(Context context) {
        final FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
        helper.validate();

        final ReadableConfig options = helper.getOptions();

        String url = options.get(URL);
        String type = options.get(TYPE);
        String header = options.get(HEADERS);
        String body = options.get(BODY);


        String unionType = options.get(UNION_TYPE);
        String unionUrl = options.get(UNION_URL);
        String unionData = options.get(UNION_DATA);

         DecodingFormat<DeserializationSchema<RowData>> decodingFormat   =helper.discoverDecodingFormat(
                DeserializationFormatFactory.class,
                FactoryUtil.FORMAT);;


        final DataType producedDataType =
                context.getCatalogTable().getResolvedSchema().toPhysicalRowDataType();


        return new HttpDynamicTableSource(type,url,header,body,decodingFormat,producedDataType,unionType,unionUrl,unionData);
    }

    @Override
    public String factoryIdentifier() {
        return IDENTIFIER;
    }

    @Override
    public Set<ConfigOption<?>> requiredOptions() {
        Set<ConfigOption<?>> requiredOptions = new HashSet();
        requiredOptions.add(URL);
        requiredOptions.add(TYPE);
        requiredOptions.add(UNION_TYPE);
        requiredOptions.add(UNION_DATA);
        requiredOptions.add(UNION_URL);
        return requiredOptions;
    }

    @Override
    public Set<ConfigOption<?>> optionalOptions() {
        Set<ConfigOption<?>> optionalOptions = new HashSet();
        optionalOptions.add(HEADERS);
        optionalOptions.add(BODY);
        optionalOptions.add(FORMAT);
        return optionalOptions;
    }


}
